package com.wsjj.yjh;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

public class MyScalarFunctionDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 2L, 2),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s3", 4L, 4)
        );

        Table table = tableEnv.fromDataStream(sensorDS);
        tableEnv.createTemporaryView("sensor",table);

        tableEnv.createTemporaryFunction("HashFunction",HashFuntion.class);

        tableEnv.sqlQuery("slect HashFunction(id) from sensor")
                .execute()
                .print();


    }


    public static class HashFuntion extends ScalarFunction{
        public int eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o){
            return o.hashCode();
        }

    }
}
